EasySpiderでCapSolverインテグレーションを使用してCaptchaを解く方法

pip install requests

import requests
import time
from typing import Optional
from dataclasses import dataclass

CAPSOLVER_API_KEY = 'YOUR_CAPSOLVER_API_KEY'

@dataclass
class TaskResult:
    status: str
    solution: Optional[dict] = None
    error_description: Optional[str] = None


class CapSolverService:
    def __init__(self, api_key: str = CAPSOLVER_API_KEY):
        self.api_key = api_key
        self.base_url = 'https://api.capsolver.com'

    def create_task(self, task_data: dict) -> str:
        response = requests.post(
            f'{self.base_url}/createTask',
            json={
                'clientKey': self.api_key,
                'task': task_data
            }
        )
        data = response.json()

        if data.get('errorId', 0) != 0:
            raise Exception(f"CapSolverエラー: {data.get('errorDescription')}")

        return data['taskId']

    def get_task_result(self, task_id: str, max_attempts: int = 60) -> TaskResult:
        for _ in range(max_attempts):
            time.sleep(2)

            response = requests.post(
                f'{self.base_url}/getTaskResult',
                json={
                    'clientKey': self.api_key,
                    'taskId': task_id
                }
            )
            data = response.json()

            if data.get('status') == 'ready':
                return TaskResult(
                    status='ready',
                    solution=data.get('solution')
                )

            if data.get('status') == 'failed':
                raise Exception(f"タスク失敗: {data.get('errorDescription')}")

        raise Exception('CAPTCHA解決待ちのタイムアウト')

    def solve_recaptcha_v2(self, website_url: str, website_key: str) -> str:
        task_id = self.create_task({
            'type': 'ReCaptchaV2TaskProxyLess',
            'websiteURL': website_url,
            'websiteKey': website_key
        })

        result = self.get_task_result(task_id)
        return result.solution.get('gRecaptchaResponse', '') if result.solution else ''

    def solve_turnstile(
        self,
        website_url: str,
        website_key: str,
        action: Optional[str] = None,
        cdata: Optional[str] = None
    ) -> str:
        task_data = {
            'type': 'AntiTurnstileTaskProxyLess',
            'websiteURL': website_url,
            'websiteKey': website_key
        }

        if action or cdata:
            metadata = {}
            if action:
                metadata['action'] = action
            if cdata:
                metadata['cdata'] = cdata
            task_data['metadata'] = metadata

        task_id = self.create_task(task_data)
        result = self.get_task_result(task_id)
        return result.solution.get('token', '') if result.solution else ''


# グローバルインスタンス
capsolver = CapSolverService()

// reCAPTCHA要素からsitekeyを抽出
var siteKey = document.querySelector('.g-recaptcha').getAttribute('data-sitekey');
var pageUrl = window.location.href;

// 次のステップで使用するための保存
window.captchaSiteKey = siteKey;
window.captchaPageUrl = pageUrl;

// デバッグ用のログ
console.log('Site Key:', siteKey);
console.log('Page URL:', pageUrl);

// 外部スクリプトから渡されたトークン
var token = arguments[0];  // EasySpiderから渡される

// レスポンステキストエリアを表示しトークンを設定
document.getElementById('g-recaptcha-response').style.display = 'block';
document.getElementById('g-recaptcha-response').value = token;

// コールバック関数がある場合、実行
if (typeof ___grecaptcha_cfg !== 'undefined') {
    var clients = ___grecaptcha_cfg.clients;
    if (clients) {
        Object.keys(clients).forEach(function(key) {
            var client = clients[key];
            if (client.callback) {
                client.callback(token);
            }
        });
    }
}

// ノードタイトル: TS_SiteKey
return document.querySelector('.cf-turnstile')?.getAttribute('data-sitekey') || '';

// ノードタイトル: TS_Action（オプション）
return document.querySelector('.cf-turnstile')?.getAttribute('data-action') || '';

// ノードタイトル: TS_Cdata（オプション）
return document.querySelector('.cf-turnstile')?.getAttribute('data-cdata') || '';

python captcha_solver.py "{{current_url}}" "Field[\"TS_SiteKey\"]" "turnstile" "Field[\"TS_Action\"]" "Field[\"TS_Cdata\"]"

var token = `Field["Token"]`;
var inputs = document.querySelectorAll('input[name="cf-turnstile-response"], input[name="turnstile-response"]');
inputs.forEach(function(el) {
    el.value = token;
    el.dispatchEvent(new Event('input', { bubbles: true }));
    el.dispatchEvent(new Event('change', { bubbles: true }));
});

# easyspider_executestage.pyの先頭にインポートを追加
import requests
import time

# CapSolverクラスを追加
class CapSolverService:
    # ...（上記の実装）...
    pass

# グローバルインスタンスを作成
capsolver = CapSolverService('YOUR_CAPSOLVER_API_KEY')

# これにより、eval/execブロック内でcapsolver.solve_recaptcha_v2()が利用可能になります

# EasySpiderのPython実行ブロックで
site_key = driver.find_element_by_class_name('g-recaptcha').get_attribute('data-sitekey')
token = capsolver.solve_recaptcha_v2(current_url, site_key)
driver.execute_script(f'''
    document.getElementById('g-recaptcha-response').value = `{token}`;
''')

import time
from functools import wraps
from typing import Callable, TypeVar, Any

T = TypeVar('T')

def retry(max_retries: int = 3, exponential_backoff: bool = True):
    """指数バックオフによるリトライロジックのデコレータ。"""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> T:
            last_exception = None

            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e

                    if attempt < max_retries - 1:
                        delay = (2 ** attempt) if exponential_backoff else 1
                        print(f'試行 {attempt + 1} が失敗しました。{delay}秒後に再試行します...')
                        time.sleep(delay)

            raise last_exception

        return wrapper
    return decorator


@retry(max_retries=3, exponential_backoff=True)
def solve_with_retry(capsolver, url: str, site_key: str) -> str:
    return capsolver.solve_recaptcha_v2(url, site_key)

def check_balance(api_key: str) -> float:
    response = requests.post(
        'https://api.capsolver.com/getBalance',
        json={'clientKey': api_key}
    )
    data = response.json()
    return data.get('balance', 0)


def main():
    balance = check_balance(CAPSOLVER_API_KEY)

    if balance < 1:
        print('警告: CapSolverの残高が少ないです！再充電してください。')
    else:
        print(f'現在の残高: ${balance:.2f}')

# 基本的な実行
python easyspider_executestage.py --id [task_id] --read_type local --headless 1

# 全パラメータで実行
python easyspider_executestage.py \
    --ids [0] \
    --server_address http://localhost:8074 \
    --config_folder "./" \
    --headless 1 \
    --read_type local

#!/usr/bin/env python3
"""
CapSolver ヘルパー スクリプト for EasySpider
使用方法: python captcha_solver.py <url> <site_key> <captcha_type> [action] [cdata]
"""

import sys
import requests
import time

CAPSOLVER_API_KEY = 'YOUR_CAPSOLVER_API_KEY'

def solve_captcha(url, site_key, captcha_type='recaptcha_v2', action=None, cdata=None):
    """CAPTCHA を解決し、トークンを返します。"""

    if captcha_type == 'recaptcha_v2':
        task_type = 'ReCaptchaV2TaskProxyLess'
    elif captcha_type == 'turnstile':
        task_type = 'AntiTurnstileTaskProxyLess'
    else:
        raise ValueError(f'不明な CAPTCHA タイプ: {captcha_type}')

    # タスクを作成
    task = {
        'type': task_type,
        'websiteURL': url,
        'websiteKey': site_key
    }
    if captcha_type == 'turnstile' and (action or cdata):
        metadata = {}
        if action:
            metadata['action'] = action
        if cdata:
            metadata['cdata'] = cdata
        task['metadata'] = metadata

    response = requests.post(
        'https://api.capsolver.com/createTask',
        json={
            'clientKey': CAPSOLVER_API_KEY,
            'task': task
        }
    )
    data = response.json()

    if data.get('errorId', 0) != 0:
        raise Exception(f"エラー: {data.get('errorDescription')}")

    task_id = data['taskId']

    # 結果をポーリング
    for _ in range(60):
        time.sleep(2)

        response = requests.post(
            'https://api.capsolver.com/getTaskResult',
            json={
                'clientKey': CAPSOLVER_API_KEY,
                'taskId': task_id
            }
        )
        data = response.json()

        if data.get('status') == 'ready':
            solution = data.get('solution', {})
            return solution.get('gRecaptchaResponse') or solution.get('token')

        if data.get('status') == 'failed':
            raise Exception(f"失敗: {data.get('errorDescription')}")

    raise Exception('タイムアウト')


if __name__ == '__main__':
    if len(sys.argv) < 3:
        print('使用方法: python captcha_solver.py <url> <site_key> [captcha_type] [action] [cdata]')
        sys.exit(1)

    url = sys.argv[1]
    site_key = sys.argv[2]
    captcha_type = sys.argv[3] if len(sys.argv) > 3 else 'recaptcha_v2'
    action = sys.argv[4] if len(sys.argv) > 4 else None
    cdata = sys.argv[5] if len(sys.argv) > 5 else None

    try:
        token = solve_captcha(url, site_key, captcha_type, action, cdata)
        print(token)  # EasySpider がキャプチャするためのトークンを出力
    except Exception as e:
        print(f'エラー: {e}', file=sys.stderr)
        sys.exit(1)

python captcha_solver.py "{{current_url}}" "{{site_key}}" "recaptcha_v2"

python captcha_solver.py "{{current_url}}" "{{site_key}}" "turnstile" "{{action}}" "{{cdata}}"